package org.raftkv.raft.core.storage.impl;

import org.raftkv.commons.Tuple;
import org.raftkv.raft.core.storage.Storage;
import org.raftkv.raft.core.storage.exceptions.CompactedException;
import org.raftkv.raft.core.storage.exceptions.SnapOutOfDateException;
import org.raftkv.raft.core.storage.exceptions.UnavailableException;
import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.ConfState;
import org.raftkv.raft.pb.Entry;
import org.raftkv.raft.pb.HardState;
import org.raftkv.raft.pb.Snapshot;
import org.raftkv.raft.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * Storage 内存存储实现
 *
 * @author wzy
 */
public class MemoryStorage implements Storage {

    private static final Logger LOGGER = LoggerFactory.getLogger(MemoryStorage.class);

    private HardState hardState;

    private Snapshot snapshot;

    private List<Entry> entries;

    public MemoryStorage() {
        hardState = new HardState();
        snapshot = new Snapshot();
        entries = new ArrayList<>();
        entries.add(new Entry());
    }

    public MemoryStorage(HardState hardState, List<Entry> entries) {
        this.hardState = hardState;
        this.entries = entries;
    }

    public MemoryStorage(List<Entry> entries) {
        this.entries = new ArrayList<>(entries);
    }

    public static MemoryStorage newMemoryStorage() {
        MemoryStorage storage = new MemoryStorage();
        storage.entries = new ArrayList<>();
        storage.entries.add(new Entry());
        return storage;
    }

    @Override
    public Tuple<HardState, ConfState> initialState() {
        return new Tuple<>(hardState, snapshot.getMetadata().getConfState());
    }

    /**
     * 保存现在的 HardState
     */
    public synchronized void setHardState(HardState hardState) {
        this.hardState = hardState;
    }

    @Override
    public synchronized List<Entry> entries(long lo, long hi, long maxSize) {
        long offset = entries.get(0).getIndex();
        if (lo <= offset) {
            throw new CompactedException();
        }
        if (hi > lastIndex() + 1) {
            String errMsg = String.format("entries' hi(%d) is out of bound lastindex(%d)", hi, lastIndex());
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
        // 只包含虚拟 entry
        if (entries.size() == 1) {
            throw new UnavailableException();
        }

        List<Entry> ents = entries.subList((int) (lo - offset), (int) (hi - offset));
        return Utils.limitSize(ents, maxSize);
    }

    @Override
    public synchronized long term(long index) {
        long offset = entries.get(0).getIndex();
        // 如果比当前数据最小的索引还小，说明已经被 compact 过了
        if (index < offset) {
            throw new CompactedException();
        }
        // 如果超过当前数组大小，返回不可用
        if (index - offset >= entries.size()) {
            throw new UnavailableException();
        }
        return entries.get((int) (index - offset)).getTerm();
    }

    @Override
    public synchronized long lastIndex() {
        return lastIndexInternal();
    }

    private long lastIndexInternal() {
        return entries.get(0).getIndex() + entries.size() - 1;
    }

    @Override
    public synchronized long firstIndex() {
        return firstIndexInternal();
    }

    private long firstIndexInternal() {
        return entries.get(0).getIndex() + 1;
    }

    @Override
    public synchronized Snapshot snapshot() {
        return snapshot;
    }

    /**
     * 该函数用给定的快照内容覆盖此 Storage 的内容
     */
    public synchronized void applySnapshot(Snapshot snapshot) {
        long msIndex = this.snapshot.getMetadata().getIndex();
        long snapIndex = snapshot.getMetadata().getIndex();
        if (msIndex >= snapIndex) {
            throw new SnapOutOfDateException();
        }
        this.snapshot = snapshot;
        this.entries = new ArrayList<>();
        // 插入一条空数据
        this.entries.add(new Entry(snapshot.getMetadata().getIndex(), snapshot.getMetadata().getTerm()));
    }

    /**
     * 该函数制作一个快照，可以使用 snapshot() 检索该快照，并可用于重建该点的状态。
     * <br/>
     * 如果自上次压缩以来进行了任何配置更改，则必须传入上次 ApplyConfChange 的结果。
     */
    public synchronized Snapshot createSnapshot(long index, ConfState cs, byte[] data) {
        if (index <= snapshot.getMetadata().getIndex()) {
            throw new SnapOutOfDateException();
        }

        long offset = entries.get(0).getIndex();
        if (index > lastIndexInternal()) {
            String errMsg = String.format("snapshot %d is out of bound lastindex(%d)", index, lastIndexInternal());
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }

        // 更新快照中的数据
        snapshot.getMetadata().setIndex(index);
        snapshot.getMetadata().setTerm(entries.get((int) (index - offset)).getTerm());
        if (cs != null) {
            snapshot.getMetadata().setConfState(cs);
        }
        snapshot.setData(data);
        return snapshot;
    }

    /**
     * 该函数会丢弃 compactIndex 之前的所有日志 entry。
     * <br/>
     * 应用程序不应该尝试压缩大于 raftLog.applied 的 entry。
     */
    public synchronized void compact(long compactIndex) {
        long offset = entries.get(0).getIndex();
        if (compactIndex <= offset) {
            throw new CompactedException();
        }
        if (compactIndex > lastIndexInternal()) {
            String errMsg = String.format("compact %d is out of bound lastindex(%d)", compactIndex, lastIndexInternal());
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }

        long i = compactIndex - offset;
        List<Entry> ents = new ArrayList<>((int) (1 + entries.size() - i));
        Entry firstEntry = entries.get((int) i);
        ents.add(new Entry(firstEntry.getIndex(), firstEntry.getTerm()));
        ents.addAll(entries.subList((int) i + 1, entries.size()));
        entries = ents;
    }

    /**
     * 追加新条目到存储中
     */
    public synchronized void append(List<Entry> entries) {
        if (entries.size() == 0) {
            return;
        }
        long first = firstIndexInternal();
        long last = entries.get(0).getIndex() + entries.size() - 1;

        // 如果没有新条目，直接返回
        if (last < first) {
            return;
        }
        // 如果当前已经包含传入数据中的一部分，那么已经有的那部分数据可以不用重复添加进来
        if (first > entries.get(0).getIndex()) {
            entries = entries.subList((int) (first - entries.get(0).getIndex()), entries.size());
        }

        // 计算传入数据到当前已经保留数据的偏移量
        long offset = entries.get(0).getIndex() - this.entries.get(0).getIndex();
        if (this.entries.size() > offset) {
            // 如果当前数据量大于偏移量，说明 offset 之前的数据从原有数据中取得，之后的数据从传入的数据中取得
            this.entries = new ArrayList<>(this.entries.subList(0, (int) offset));
            this.entries.addAll(entries);
        } else if (this.entries.size() == offset) {
            // offset 刚好等于当前数据量，说明传入的数据刚好紧挨着当前的数据，所以直接添加进来就好了
            this.entries.addAll(entries);
        } else {
            String errMsg = String.format("missing log entry [last: %d, append at: %d]", lastIndexInternal(), entries.get(0).getIndex());
            LOGGER.error(errMsg);
            throw new RaftException(errMsg);
        }
    }

    public HardState getHardState() {
        return hardState;
    }

    public Snapshot getSnapshot() {
        return snapshot;
    }

    public List<Entry> getEntries() {
        return entries;
    }

}
